package com.niit.workcoun;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/**
 *
 * @desc:本类就是MapReduce程序中Reduce阶段的处理类对应都ReduceTask
 *
 * KEYIN:表示的是reduce 阶段输入kV水的类型对应着map的输出的key因此本需求中就是单词Text
 * VALUEIN:表示的是reduce阶段输Akv中v的类型对应者map的输出的value因此本需求中就是单词次数1 LongWritable
 * KEYOUT:表示的是reduce 阶段输出kv中k的类型跟业务相关 本需求中还是单词 Text
 * VALUEOUT:表示的是reduce阶段输出kv中v的类型跟业务相关 本需求中还是单词总次数 LongWritable
 */
public class WordCountReducer extends Reducer<Text, LongWritable,Text,LongWritable> {


    /**
     * todo Q;当map的所有输出数据来到reduce之后该如何调用reduce方法进行处理呢?
     *
            <hello, 1><hadoop, 1><hello, 1><hello, 1><hadoop,1>
      1.排序I规则:根据key的字典序进行排序a-z
                <hadoop, 1><hadoop, 1><hadoop, 1><hello, 1><hello,1>
     * 2.分组规则: key相同的分为一组
     *
             <hadoop, 1><hadoop, 1><hadoop, 1>
             <hello, 1><hello,1>
     * 3、分组之后，同一组的数据组成一个新的kv键值对，调用一次reduce方法。reduce方法基于分组调用的一个分组调用一次。
     *
            todo同-组中数据组成- -个新的kv键值对。
            新key:该组共同的key
            新value:该组所有的value组成的一个送代器Iterable
             <hadoop, 1><hadoop , 1><hadoop, 1----><hadoop, Iterable[1,1,1]>
             <hello, 1><hello,1>----> <hello, Iterable[1,1]>
     */


    @Override
    protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
        //定义次数结果。
        LongWritable result = new LongWritable();

        //统计变量
        long count = 0;
        //遍历一组数据，取出该组所有的value  <hadoop, Iterable[1,1,1]>
        for (LongWritable value : values){
            // 所有的value累加 就是该单词的总次数
            count += value.get(); // 0+1 + 1 + 1 = 3
        }
        result.set(count);
        //输出最终结果<单词，总次数>
        //所有的输出，按照key的字典升序进行排序（默认）
        context.write(key,result);

    }
}
